package com.wsj.spring.kafka.kafkademo.consumer;

import com.wsj.spring.kafka.kafkademo.consts.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * 接收消息
 *
 * @author WangSongJun
 * @date 2018-06-30 21:24 星期六
 */
@Slf4j
@Component
public class KafkaReceiver {

    @KafkaListener(topics={Constants.TOPIC_1})
    public void listen(ConsumerRecord record) {
        log.info("record:{}", record);
        Optional kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            log.info("receive message:{}",kafkaMessage.get());
        }
    }
}
